/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.parser.mp.handler;

import static cn.ac.iie.di.parser.mp.main.ParserMpMain.localFileSize;
import cn.ac.iie.di.parser.mp.utils.FileTools;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class ParserFileHandler implements Runnable {

    static private final Logger logger = Logger.getLogger(ParserFileHandler.class.getName());

    private String localpath = "";
    private ArrayBlockingQueue<String> dataBlockingQueue = null;

    public ParserFileHandler(String localpath, ArrayBlockingQueue dataBlockingQueue) {
        this.localpath = localpath;
        this.dataBlockingQueue = dataBlockingQueue;
    }

    @Override
    public void run() {
        logger.info("the parser file path is " + localpath);
        while (true) {
            try {
                File localFile = new File(localpath);
                if (localFile.exists()) {
                    File[] localFilelist = localFile.listFiles();
                    int length = localFilelist.length;
//                    logger.info(localpath + " has " + length + " files. And the LOCALFILESIZE is : " + localFileSize);

                    List<File> okList = new ArrayList<File>();
                    List<File> fileList = new ArrayList<File>();
                    List<File> finalList = new ArrayList<File>();//最终解析列表

                    for (int i = 0; i < length; i++) {
                        if (localFilelist[i].getName().endsWith(".ok")) {
                            okList.add(localFilelist[i]);
                        } else {
                            fileList.add(localFilelist[i]);
                        }
                    }

                    boolean flag = true;
                    int count = 0;
                    for (int i = 0; i < fileList.size(); i++) {
                        if (flag) {
                            for (int j = 0; j < okList.size(); j++) {
                                if ((fileList.get(i).getName() + ".ok").equals(okList.get(j).getName())) {
                                    finalList.add(fileList.get(i));
                                    count++;
                                }
                                if (count == localFileSize) {
                                    flag = false;
                                }
                            }
                        }
                        if (!flag) {
                            break;
                        }
                    }

                    //打印文件的数目
                    logger.info(localpath + " has " + finalList.size() + " files. And the LOCALFILESIZE is : " + localFileSize);

                    if (!finalList.isEmpty()) {
                        for (File file : finalList) {
                            readFileByLines(file);//解析文件
                            logger.info("file name is : " + file.getName() + "  paser sucessfully ");
                        }
                        Thread.sleep(3000);
                    } else {
                        logger.info("the localPath---[" + localpath + "] doesn't hava available file, and sleep 3 seconds.");
                        Thread.sleep(3000);
                    }
                } else {
                    logger.info(localpath + " do not exist.");
                    Thread.sleep(3000);
                }
            } catch (Exception ex) {
                logger.error("mathch files failed, cause " + ex.getMessage(), ex);
            }
        }
    }

    public void readFileByLines(File file) {
        BufferedReader reader = null;
        try {
            long begin = System.currentTimeMillis();
            int sum = 0;
            reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), "UTF-8"));
            String tempString = null;
            while ((tempString = reader.readLine()) != null) {
                try {
                    while (!dataBlockingQueue.offer(tempString)) {
                        Thread.sleep(1000);
                    }
                    sum++;
                } catch (InterruptedException ex) {
                    logger.error("add line into " + dataBlockingQueue + " failed, cause: " + ex.getMessage(), ex);
                }
            }
            new FileTools().moveFileToNewPath(file.getName(), localpath);
            long end = System.currentTimeMillis();
            logger.info("readFileByLines--- " + file + " :q " + (end - begin) + " ms, and has records " + sum);
        } catch (Exception ex) {
            logger.error("read " + file + " by line failed, cause: " + ex.getMessage(), ex);
        } finally {
            try {
                reader.close();
            } catch (Exception e) {
                logger.error("close BufferedReader failed, cause: " + e.getMessage(), e);
            }
        }
    }
}
